package com.doubao.chat.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.doubao.chat.dto.ChatMessageDTO;
import com.doubao.chat.dto.MessageProviderInfo;
import com.doubao.chat.dto.VoiceConfigDTO;
import com.doubao.chat.dto.request.*;
import com.doubao.chat.entity.ChatMessage;
import com.doubao.chat.entity.ChatSession;
import com.doubao.chat.mapper.ChatMessageMapper;
import com.doubao.chat.mapper.ChatSessionMapper;
import com.doubao.chat.service.AIProviderService;
import com.doubao.chat.service.ChatMessageService;
import com.doubao.common.exception.BusinessException;
import com.doubao.common.result.Result;
import com.doubao.common.utils.RedisTemplateFactory;
import com.doubao.device.entity.SmartDevice;
import com.doubao.device.service.DeviceVoiceService;
import com.doubao.device.service.SmartDeviceService;
import com.doubao.user.entity.User;
import com.doubao.user.mapper.UserMapper;
import com.doubao.user.service.MembershipService;
import com.doubao.user.service.TokenService;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.DigestUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 聊天消息服务实现类
 */
@Service
@Slf4j
public class ChatMessageServiceImpl extends ServiceImpl<ChatMessageMapper, ChatMessage> implements ChatMessageService {
    @Autowired
    private ChatSessionMapper sessionMapper;

    @Autowired
    private SmartDeviceService deviceService;

    @Autowired
    @Qualifier("AIProviderServiceImpl") // 使用类名作为bean名
    private AIProviderService aiProviderService;

    @Autowired
    private DeviceVoiceService voiceService;

    @Autowired
    private TokenService tokenService;

    // 存储SSE连接的映射
    private final ConcurrentHashMap<String, SseEmitter> sseEmitters = new ConcurrentHashMap<>();

    // 心跳检测线程池
    private final ScheduledExecutorService heartbeatExecutor = Executors.newScheduledThreadPool(1);

    @Value("${chat.context-message-limit:10}")
    private Integer contextMessageLimit;

    @Value("${chat.stream-timeout:60000}")
    private Long streamTimeout;

    @Value("${chat.stream-heartbeat:15000}")
    private Long streamHeartbeat;

    @Value("${chat.session-reuse-hours:24}")
    private int sessionReuseHours;

    @Value("${chat.default-tokens-enabled:true}")
    private boolean defaultTokensEnabled;

    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private MembershipService membershipService;

    @PostConstruct
    public void init() {
        this.redisTemplate = RedisTemplateFactory.getRedisTemplate();
        log.info("ChatMessageServiceImpl使用RedisTemplateFactory获取RedisTemplate");
    }

    @Autowired
    private ApplicationContext applicationContext;

    @Autowired
    private UserMapper userMapper;

    /**
     * 本地检查用户token余额
     *
     * @param userId         用户ID
     * @param requiredTokens 需要的token数量
     * @return 是否有足够的token
     */
    private boolean checkLocalTokenBalance(Long userId, int requiredTokens) {
        try {
            User user = userMapper.selectById(userId);
            if (user == null) {
                log.warn("用户不存在: userId={}", userId);
                return false;
            }

            Integer balance = user.getTotalTokensBalance();
            if (balance == null) {
                log.warn("用户token余额为null: userId={}", userId);
                return false;
            }

            boolean hasEnough = balance >= requiredTokens;
            log.info("本地检查用户token余额: userId={}, 余额={}, 需要={}, 结果={}",
                    userId, balance, requiredTokens, hasEnough ? "足够" : "不足");
            return hasEnough;
        } catch (Exception e) {
            log.error("本地检查用户token余额异常: {}", e.getMessage(), e);
            return false; // 发生异常时返回false，保守处理
        }
    }

    /**
     * 在本地记录 token 使用情况
     * 使用 Redis 存储以确保高可用性和持久性
     */
    private void saveLocalTokenUsage(Long userId, int baseTokens) {
        try {
            // Redis key format for daily usage tracking
            String currentDate = java.time.LocalDate.now().format(java.time.format.DateTimeFormatter.ofPattern("yyyyMMdd"));
            String redisKey = "user_token_usage:" + userId + ":" + currentDate;

            // Record the token usage
            redisTemplate.opsForValue().increment(redisKey, baseTokens);
            redisTemplate.expire(redisKey, 7, java.util.concurrent.TimeUnit.DAYS);

            // Also record this in a "failed deductions" queue for later processing
            String pendingDeductionsKey = "pending_token_deductions";
            Map<String, Object> deductionRecord = new HashMap<>();
            deductionRecord.put("userId", userId);
            deductionRecord.put("tokens", baseTokens);
            deductionRecord.put("timestamp", System.currentTimeMillis());
            redisTemplate.opsForList().rightPush(pendingDeductionsKey, JSON.toJSONString(deductionRecord));

            log.info("记录本地token使用和待处理扣减: userId={}, tokens={}", userId, baseTokens);
        } catch (Exception e) {
            log.error("保存本地token使用记录失败: {}", e.getMessage(), e);
        }
    }

    // 3. 在 processPendingTokenDeductions 方法中添加详细日志
    @Scheduled(cron = "0 0 * * * *") // 每小时整点执行
    public void processPendingTokenDeductions() {
        log.info("开始处理待处理的 token 扣减记录...");
        String pendingDeductionsKey = "pending_token_deductions";
        int processedCount = 0;
        int failedCount = 0;

        while (true) {
            try {
                String deductionJson = (String) redisTemplate.opsForList().leftPop(pendingDeductionsKey);
                if (deductionJson == null) {
                    break; // No more pending deductions
                }

                Map<String, Object> deduction = JSON.parseObject(deductionJson);
                Long userId = Long.valueOf(deduction.get("userId").toString());
                Integer tokens = Integer.valueOf(deduction.get("tokens").toString());

                // 直接使用本地 TokenService 进行扣减
                try {
                    Result<Void> result = tokenService.preDeductTokens(userId, tokens);
                    if (result.isSuccess()) {
                        log.info("成功处理 token 扣减: userId={}, tokens={}", userId, tokens);
                        processedCount++;
                    } else {
                        // 如果失败，推回队列
                        redisTemplate.opsForList().rightPush(pendingDeductionsKey, deductionJson);
                        log.warn("处理 token 扣减失败: userId={}, tokens={}, 错误: {}",
                                userId, tokens, result.getMessage());
                        failedCount++;
                        break; // 停止处理，避免无限循环处理问题记录
                    }
                } catch (Exception e) {
                    // 如果通信错误，推回队列
                    redisTemplate.opsForList().rightPush(pendingDeductionsKey, deductionJson);
                    log.error("调用 token 服务异常: userId={}, tokens={}, 错误: {}",
                            userId, tokens, e.getMessage());
                    failedCount++;
                    break;
                }
            } catch (Exception e) {
                log.error("处理 token 扣减记录时发生异常: {}", e.getMessage(), e);
                failedCount++;
            }
        }

        log.info("待处理 token 扣减处理完成: 成功={}, 失败={}", processedCount, failedCount);
    }

    @Override
    @Transactional
    public Result<ChatMessageDTO> sendMessage(Long userId, SendMessageRequest request) {
        if (userId == null) {
            return Result.failed("用户未登录");
        }

        if (!StringUtils.hasText(request.getContent())) {
            return Result.failed("消息内容不能为空");
        }

//        // 检查用户聊天次数是否足够
//        boolean hasSufficientChatCount = membershipService.hasSufficientChatCount(userId);
//        if (!hasSufficientChatCount) {
//            return Result.failed("今日聊天次数已用完，请升级会员或明天再来");
//        }

        // 每条消息需要的token数量
        int requiredTokens = 10;

        // 检查用户token余额是否足够 - 直接调用本地服务
        boolean hasEnoughTokens = tokenService.hasEnoughTokensForMessage(userId);
        if (!hasEnoughTokens) {
            return Result.failed("Token余额不足，请充值后再发送消息");
        }

        // 预先扣减token - 直接调用本地服务
        Result<Void> deductResult = tokenService.preDeductTokens(userId, requiredTokens);
        if (!deductResult.isSuccess()) {
            // 如果是余额不足导致的失败
            if (deductResult.getMessage() != null && deductResult.getMessage().contains("余额不足")) {
                return Result.failed(deductResult.getMessage());
            }
            // 如果是其他原因导致的失败，返回错误信息
            return Result.failed("预扣减 Token 失败: " + deductResult.getMessage());
        }

//        // 消费聊天次数
//        Result<Void> chatResult = membershipService.consumeChatCount(userId);
//        if (!chatResult.isSuccess()) {
//            log.warn("消费聊天次数失败，但允许继续: {}", chatResult.getMessage());
//            // 允许继续，不阻断流程
//        }

        // 获取或创建会话
        ChatSession session;
        if (request.getSessionId() != null) {
            session = getSessionById(userId, request.getSessionId());
            if (session == null) {
                return Result.failed("会话不存在或无访问权限");
            }
        } else {
            // 查找最近的活跃会话，如果有则复用
            session = findRecentActiveSession(userId, request.getDeviceId());

            // 如果没有找到活跃会话，则创建新会话
            if (session == null) {
                session = createSession(userId, request.getDeviceId());
            } else {
                log.info("复用现有会话: {}", session.getId());
            }
        }

        // 检查智能体是否存在
        SmartDevice device = deviceService.getById(request.getDeviceId());
        if (device == null || device.getStatus() != 1) {
            return Result.failed("智能体不存在或已禁用");
        }

        // 保存用户消息
        ChatMessage userMessage = saveUserMessage(userId, session.getId(), device.getId(), request.getContent(), request.getClientMsgId());

        // 获取AI提供商信息
        Result<MessageProviderInfo> providerResult = aiProviderService.getProviderInfo(device.getId());
        if (!providerResult.isSuccess()) {
            return Result.failed(providerResult.getMessage());
        }
        MessageProviderInfo providerInfo = providerResult.getData();

        // 获取系统提示词
        Result<String> promptResult = aiProviderService.getSystemPrompt(device.getId());
        String systemPrompt = promptResult.isSuccess() ? promptResult.getData() : "";

        // 获取上下文消息
        List<ChatMessage> contextMessages = getContextMessages(session.getId(),
                device.getMaxContextLength() != null ? device.getMaxContextLength() : contextMessageLimit);

        // 向AI提供商发送消息并获取回复
        Result<String> replyResult = aiProviderService.sendMessageToProvider(
                device.getId(), systemPrompt, contextMessages, request.getContent());

        if (!replyResult.isSuccess()) {
            return Result.failed(replyResult.getMessage());
        }

        // 保存AI回复消息
        ChatMessage aiMessage = saveAssistantMessage(userId, session.getId(), device.getId(),
                replyResult.getData(), providerInfo.getProviderId(), providerInfo.getModelId());

        // 更新会话信息
        updateSessionAfterMessage(session.getId());

        // 增加智能体使用次数
        incrementDeviceUsageCount(device.getId());

        // 如果是新会话且有之前的会话ID，则归档之前的会话
        if (request.getPreviousSessionId() != null && !request.getPreviousSessionId().equals(session.getId())) {
            // 归档逻辑（可选）
            // chatArchiveService.archiveSessionToTxt(request.getPreviousSessionId(), userId);
        }

        // 转换为DTO返回
        ChatMessageDTO messageDTO = new ChatMessageDTO();
        BeanUtils.copyProperties(aiMessage, messageDTO);
        messageDTO.setIsCurrentUser(false);

        // 添加会话ID到响应中
        messageDTO.setSessionId(session.getId());

        return Result.success(messageDTO);
    }

    @Override
    public SseEmitter sendMessageStream(Long userId, SendMessageRequest request) {
        if (userId == null) {
            throw new BusinessException("用户未登录");
        }

        if (!StringUtils.hasText(request.getContent())) {
            throw new BusinessException("消息内容不能为空");
        }

//        // 检查用户聊天次数是否足够
//        boolean hasSufficientChatCount = membershipService.hasSufficientChatCount(userId);
//        if (!hasSufficientChatCount) {
//            throw new BusinessException("今日聊天次数已用完，请升级会员或明天再来");
//        }

        // 检查用户token余额是否足够
        boolean hasEnoughTokens = tokenService.hasEnoughTokensForMessage(userId);
        if (!hasEnoughTokens) {
            throw new BusinessException("Token余额不足，请充值后再发送消息");
        }

        // 扣减基础token费用
        Result<Void> deductResult = tokenService.preDeductTokens(userId, 50);
        if (!deductResult.isSuccess()) {
            throw new BusinessException(deductResult.getMessage());
        }

//        // 消费聊天次数
//        Result<Void> chatResult = membershipService.consumeChatCount(userId);
//        if (!chatResult.isSuccess()) {
//            log.warn("消费聊天次数失败，但允许继续: {}", chatResult.getMessage());
//            // 允许继续，不阻断流程
//        }

        // 创建SSE发射器
        SseEmitter emitter = new SseEmitter(streamTimeout);
        String emitterId = UUID.randomUUID().toString();
        sseEmitters.put(emitterId, emitter);

        // 设置SSE回调
        emitter.onCompletion(() -> sseEmitters.remove(emitterId));
        emitter.onTimeout(() -> sseEmitters.remove(emitterId));
        emitter.onError(e -> {
            log.error("SSE错误: {}", e.getMessage());
            sseEmitters.remove(emitterId);
        });

        // 在单独的线程中处理消息发送
        new Thread(() -> {
            try {
                // 发送初始事件
                sendSseEvent(emitter, "start", "开始生成回复...");

                // 获取或创建会话
                ChatSession session;
                if (request.getSessionId() != null) {
                    session = getSessionById(userId, request.getSessionId());
                    if (session == null) {
                        throw new BusinessException("会话不存在或无访问权限");
                    }
                } else {
                    // 查找最近的活跃会话，如果有则复用
                    session = findRecentActiveSession(userId, request.getDeviceId());

                    // 如果没有找到活跃会话，则创建新会话
                    if (session == null) {
                        session = createSession(userId, request.getDeviceId());
                    } else {
                        log.info("流式响应复用现有会话: {}", session.getId());
                    }
                }

                // 检查智能体是否存在
                SmartDevice device = deviceService.getById(request.getDeviceId());
                if (device == null || device.getStatus() != 1) {
                    throw new BusinessException("智能体不存在或已禁用");
                }

                // 保存用户消息
                ChatMessage userMessage = saveUserMessage(userId, session.getId(), device.getId(),
                        request.getContent(), request.getClientMsgId());

                // 发送用户消息事件
                ChatMessageDTO userMessageDTO = new ChatMessageDTO();
                BeanUtils.copyProperties(userMessage, userMessageDTO);
                userMessageDTO.setIsCurrentUser(true);
                sendSseEvent(emitter, "user_message", userMessageDTO);

                // 获取AI提供商信息
                Result<MessageProviderInfo> providerResult = aiProviderService.getProviderInfo(device.getId());
                if (!providerResult.isSuccess()) {
                    throw new BusinessException(providerResult.getMessage());
                }
                MessageProviderInfo providerInfo = providerResult.getData();

                // 获取系统提示词
                Result<String> promptResult = aiProviderService.getSystemPrompt(device.getId());
                String systemPrompt = promptResult.isSuccess() ? promptResult.getData() : "";

                // 获取上下文消息
                List<ChatMessage> contextMessages = getContextMessages(session.getId(),
                        device.getMaxContextLength() != null ? device.getMaxContextLength() : contextMessageLimit);

                // 向AI提供商发送消息并获取回复
                Result<String> replyResult = aiProviderService.sendMessageToProvider(
                        device.getId(), systemPrompt, contextMessages, request.getContent());

                if (!replyResult.isSuccess()) {
                    throw new BusinessException(replyResult.getMessage());
                }

                // 模拟流式响应，将回复分割成多段
                String fullReply = replyResult.getData();
                simulateStreamingResponse(emitter, fullReply);

                // 保存AI回复消息
                ChatMessage aiMessage = saveAssistantMessage(userId, session.getId(), device.getId(),
                        fullReply, providerInfo.getProviderId(), providerInfo.getModelId());

                // 更新会话信息
                updateSessionAfterMessage(session.getId());

                // 增加智能体使用次数
                incrementDeviceUsageCount(device.getId());

                // 发送完成事件
                sendSseEvent(emitter, "done", "回复生成完成");

                // 关闭发射器
                emitter.complete();

            } catch (Exception e) {
                log.error("流式回复处理异常: {}", e.getMessage(), e);
                try {
                    sendSseEvent(emitter, "error", "处理消息出错: " + e.getMessage());
                    emitter.complete();
                } catch (Exception ex) {
                    log.error("发送错误事件失败: {}", ex.getMessage());
                }
            }
        }).start();

        // 启动心跳任务
        startHeartbeat(emitter, emitterId);

        return emitter;
    }

    @Override
    public Result<IPage<ChatMessageDTO>> getMessageList(Long userId, GetMessageListRequest request) {
        if (userId == null) {
            return Result.failed("用户未登录");
        }

        // 检查会话是否存在
        ChatSession session = getSessionById(userId, request.getSessionId());
        if (session == null) {
            return Result.failed("会话不存在或无访问权限");
        }

        // 分页查询消息
        Page<ChatMessage> page = new Page<>(request.getPage(), request.getSize());
        IPage<ChatMessage> messagePage = baseMapper.selectSessionMessages(page,
                request.getSessionId(), request.getMaxId());

        // 转换为DTO
        IPage<ChatMessageDTO> dtoPage = messagePage.convert(message -> {
            ChatMessageDTO dto = new ChatMessageDTO();
            BeanUtils.copyProperties(message, dto);
            dto.setIsCurrentUser("user".equals(message.getRole()));
            return dto;
        });

        return Result.success(dtoPage);
    }

    @Override
    @Transactional
    public Result<Void> deleteMessage(Long userId, Long messageId) {
        if (userId == null) {
            return Result.failed("用户未登录");
        }

        // 查询消息
        ChatMessage message = getById(messageId);
        if (message == null || !message.getUserId().equals(userId)) {
            return Result.failed("消息不存在或无访问权限");
        }

        // 删除消息
        removeById(messageId);

        return Result.success();
    }

    @Override
    @Transactional
    public Result<ChatMessageDTO> regenerateMessage(Long userId, Long sessionId, Long messageId) {
        if (userId == null) {
            return Result.failed("用户未登录");
        }

        // 检查会话是否存在
        ChatSession session = getSessionById(userId, sessionId);
        if (session == null) {
            return Result.failed("会话不存在或无访问权限");
        }

        // 查询消息
        ChatMessage message = getById(messageId);
        if (message == null || !message.getSessionId().equals(sessionId)) {
            return Result.failed("消息不存在或不属于该会话");
        }

        // 检查是否是assistant的消息
        if (!"assistant".equals(message.getRole())) {
            return Result.failed("只能重新生成AI的回复消息");
        }

        // 获取上下文消息(不包括当前要重新生成的消息)
        LambdaQueryWrapper<ChatMessage> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(ChatMessage::getSessionId, sessionId)
                .lt(ChatMessage::getId, messageId)
                .orderByDesc(ChatMessage::getId)
                .last("LIMIT " + contextMessageLimit);
        List<ChatMessage> contextMessages = list(queryWrapper);
        Collections.reverse(contextMessages); // 按时间升序排列

        // 找到最近一条用户消息
        ChatMessage lastUserMessage = null;
        for (int i = contextMessages.size() - 1; i >= 0; i--) {
            if ("user".equals(contextMessages.get(i).getRole())) {
                lastUserMessage = contextMessages.get(i);
                break;
            }
        }

        if (lastUserMessage == null) {
            return Result.failed("找不到用户消息");
        }

        // 获取智能体信息
        SmartDevice device = deviceService.getById(session.getDeviceId());
        if (device == null || device.getStatus() != 1) {
            return Result.failed("智能体不存在或已禁用");
        }

        // 获取AI提供商信息
        Result<MessageProviderInfo> providerResult = aiProviderService.getProviderInfo(device.getId());
        if (!providerResult.isSuccess()) {
            return Result.failed(providerResult.getMessage());
        }
        MessageProviderInfo providerInfo = providerResult.getData();

        // 获取系统提示词
        Result<String> promptResult = aiProviderService.getSystemPrompt(device.getId());
        String systemPrompt = promptResult.isSuccess() ? promptResult.getData() : "";

        // 向AI提供商发送消息并获取回复
        Result<String> replyResult = aiProviderService.sendMessageToProvider(
                device.getId(), systemPrompt, contextMessages, lastUserMessage.getContent());

        if (!replyResult.isSuccess()) {
            return Result.failed(replyResult.getMessage());
        }

        // 更新消息
        message.setContent(replyResult.getData());
        message.setUpdatedAt(LocalDateTime.now());
        updateById(message);

        // 转换为DTO返回
        ChatMessageDTO messageDTO = new ChatMessageDTO();
        BeanUtils.copyProperties(message, messageDTO);
        messageDTO.setIsCurrentUser(false);

        return Result.success(messageDTO);
    }

    @Override
    public Result<String> textToSpeech(Long userId, TextToSpeechRequest request) {
        if (userId == null) {
            return Result.failed("用户未登录");
        }

        // 检查智能体是否存在
        SmartDevice device = deviceService.getById(request.getDeviceId());
        if (device == null || device.getStatus() != 1) {
            return Result.failed("智能体不存在或已禁用");
        }

        try {
            // 获取语音配置
            Result<com.doubao.device.dto.VoiceConfigDTO> voiceResult = voiceService.getDeviceVoiceConfig(device.getId());
            if (!voiceResult.isSuccess()) {
                return Result.failed("获取语音配置失败");
            }

            com.doubao.device.dto.VoiceConfigDTO sourceConfig = voiceResult.getData();
            VoiceConfigDTO voiceConfig = new VoiceConfigDTO();
            BeanUtils.copyProperties(sourceConfig, voiceConfig);

            // 使用请求中的配置覆盖设备默认配置
            if (StringUtils.hasText(request.getVoiceType())) {
                voiceConfig.setVoiceType(request.getVoiceType());
            }

            if (request.getSpeedRatio() != null) {
                voiceConfig.setSpeedRatio(request.getSpeedRatio());
            }

            // TODO: 调用语音合成服务生成语音
            // 这里需要根据实际的语音合成服务实现
            // 示例返回一个假的URL
            return Result.success("https://example.com/tts/sample.mp3");

        } catch (Exception e) {
            log.error("文本转语音失败", e);
            return Result.failed("文本转语音失败: " + e.getMessage());
        }
    }

    /**
     * 查找最近的活跃会话
     * 如果最近24小时内有会话，则复用
     */
    private ChatSession findRecentActiveSession(Long userId, Long deviceId) {
        LambdaQueryWrapper<ChatSession> queryWrapper = new LambdaQueryWrapper<ChatSession>()
                .eq(ChatSession::getUserId, userId)
                .eq(ChatSession::getDeviceId, deviceId)
                .eq(ChatSession::getStatus, 1)  // 正常状态
                .ge(ChatSession::getLastMessageTime, LocalDateTime.now().minusHours(sessionReuseHours))
                .orderByDesc(ChatSession::getLastMessageTime)
                .last("LIMIT 1");

        return sessionMapper.selectOne(queryWrapper);
    }

    /**
     * 获取会话
     */
    private ChatSession getSessionById(Long userId, Long sessionId) {
        LambdaQueryWrapper<ChatSession> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(ChatSession::getId, sessionId)
                .eq(ChatSession::getUserId, userId)
                .eq(ChatSession::getStatus, 1); // 正常状态
        return sessionMapper.selectOne(queryWrapper);
    }

    /**
     * 创建新会话
     */
    private ChatSession createSession(Long userId, Long deviceId) {
        // 获取智能体
        SmartDevice device = deviceService.getById(deviceId);
        if (device == null || device.getStatus() != 1) {
            throw new BusinessException("智能体不存在或已禁用");
        }

        ChatSession session = new ChatSession();
        session.setUserId(userId);
        session.setDeviceId(deviceId);

        // 设置会话标题
        String title = "与" + device.getName() + "的对话";
        session.setTitle(title);

        session.setMessageCount(0);
        session.setLastMessageTime(LocalDateTime.now());
        session.setStatus(1);
        session.setCreatedAt(LocalDateTime.now());
        session.setUpdatedAt(LocalDateTime.now());

        // 保存会话
        sessionMapper.insert(session);
        log.info("创建新会话: {}", session.getId());

        return session;
    }

    /**
     * 增加智能体使用次数
     */
    private void incrementDeviceUsageCount(Long deviceId) {
        try {
            deviceService.increaseUsageCount(deviceId);
        } catch (Exception e) {
            log.error("增加智能体使用次数失败", e);
            // 非关键操作，失败不影响主流程
        }
    }

    /**
     * 保存用户消息
     */
    private ChatMessage saveUserMessage(Long userId, Long sessionId, Long deviceId,
                                        String content, String clientMsgId) {
        ChatMessage message = new ChatMessage();
        message.setUuid(UUID.randomUUID().toString());
        message.setSessionId(sessionId);
        message.setUserId(userId);
        message.setDeviceId(deviceId);
        message.setRole("user");
        message.setContent(content);
        message.setStorageType("db");
        message.setMessageHash(generateMessageHash(content));
        message.setClientMsgId(clientMsgId);
        message.setCreatedAt(LocalDateTime.now());
        message.setUpdatedAt(LocalDateTime.now());

        // 保存消息
        save(message);

        return message;
    }

    /**
     * 保存AI回复消息
     */
    private ChatMessage saveAssistantMessage(Long userId, Long sessionId, Long deviceId,
                                             String content, Long providerId, String modelId) {
        ChatMessage message = new ChatMessage();
        message.setUuid(UUID.randomUUID().toString());
        message.setSessionId(sessionId);
        message.setUserId(userId);
        message.setDeviceId(deviceId);
        message.setRole("assistant");
        message.setContent(content);
        message.setAiProviderId(providerId);
        message.setAiModelId(modelId);
        message.setStorageType("db");
        message.setMessageHash(generateMessageHash(content));
        message.setCreatedAt(LocalDateTime.now());
        message.setUpdatedAt(LocalDateTime.now());

        // 保存消息
        save(message);

        return message;
    }

    /**
     * 更新会话信息
     */
    private void updateSessionAfterMessage(Long sessionId) {
        sessionMapper.updateSessionMessageCount(sessionId);
    }

    /**
     * 获取上下文消息
     */
    private List<ChatMessage> getContextMessages(Long sessionId, Integer limit) {
        List<ChatMessage> messages = baseMapper.selectContextMessages(sessionId, limit);
        if (messages == null) {
            return new ArrayList<>();
        }
        return messages;
    }

    /**
     * 生成消息哈希
     */
    private String generateMessageHash(String content) {
        return DigestUtils.md5DigestAsHex(content.getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 发送SSE事件
     */
    private void sendSseEvent(SseEmitter emitter, String event, Object data) throws IOException {
        emitter.send(SseEmitter.event()
                .name(event)
                .data(data, MediaType.APPLICATION_JSON));
    }

    /**
     * 启动心跳任务
     */
    private void startHeartbeat(SseEmitter emitter, String emitterId) {
        heartbeatExecutor.scheduleAtFixedRate(() -> {
            try {
                if (sseEmitters.containsKey(emitterId)) {
                    emitter.send(SseEmitter.event()
                            .name("heartbeat")
                            .data("ping", MediaType.TEXT_PLAIN));
                }
            } catch (Exception e) {
                log.error("发送心跳失败: {}", e.getMessage());
                sseEmitters.remove(emitterId);
            }
        }, streamHeartbeat, streamHeartbeat, TimeUnit.MILLISECONDS);
    }

    /**
     * 模拟流式响应
     */
    private void simulateStreamingResponse(SseEmitter emitter, String fullReply) throws IOException {
        // 为了演示，将完整回复按句子分割
        String[] sentences = fullReply.split("(?<=[.!?]\\s)");
        StringBuilder currentText = new StringBuilder();

        for (String sentence : sentences) {
            currentText.append(sentence);
            sendSseEvent(emitter, "token", currentText.toString());

            // 模拟延迟
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

}